package com.dreams.dream.common.async.disruptor;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;

public class DisruptorHelper {

    private static DisruptorFactory disruptorFactory = new DisruptorFactory();

    /**
     * 
     * @param handler
     * @param eventSource
     * @param topic
     * @return
     */
    @SuppressWarnings("rawtypes")
    public static DomainMessage publish(DomainEventHandler handler, Object eventSource, String topic) {

        Disruptor disruptor = disruptorFactory.createDisruptor(handler, topic);
        
        RingBuffer ringBuffer = disruptor.getRingBuffer();

        long sequence = ringBuffer.next();
        
        DomainMessage domainMessage = new DomainMessage(eventSource);

        EventDisruptor eventDisruptor = (EventDisruptor) ringBuffer.get(sequence);
        eventDisruptor.setTopic(topic);
        eventDisruptor.setDomainMessage(domainMessage);

        ringBuffer.publish(sequence);

        return domainMessage;
    }
}
